package com.w.utils.thread.concurrent;


import org.springframework.util.Assert;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author weiwei
 * @Date 2020-11-16 21:33
 * @description 多线程提交执行器类
 **/
public class ConcurrentInvoker {

    private final ExecutorService executor;

    private String name; //执行器名称

    private Map<String, Invoker> invokers ; //具体的执行方法
    private long timeout; //活跃时间
    private boolean syncThreadLocal = false; //是否需要使用threadLocalObj
    private Object threadLocalObj; //每个线程可从threadLocal中拿到的对象

    private ConcurrentInvoker(String name,Map<String, Invoker> invokers, long timeout, int concurrentCount) {
        this.name=name;
        this.invokers = invokers;
        this.timeout = timeout;
        this.executor = new ThreadPoolExecutor(concurrentCount, concurrentCount << 1,
                30L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1 << 10),
                new CustomerThreadFactory("concurrent-invoker", true));
    }

    private ConcurrentInvoker(String name,Map<String, Invoker> invokers, long timeout, boolean syncThreadLocal, Object threadLocalObj, int concurrentCount) {
        this.name=name;
        this.invokers = invokers;
        this.timeout = timeout;
        this.syncThreadLocal = syncThreadLocal;
        this.threadLocalObj = threadLocalObj;
        this.executor = new ThreadPoolExecutor(concurrentCount, concurrentCount << 1,
                30L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1 << 10),
                new CustomerThreadFactory("concurrent-invoker", true));
    }

    public <T> T invoke(Merger<T> merger) {
        Assert.notNull(merger, "merger can not be null");
        return merger.merge(blockingInvoke());
    }


    public Map<String, Object> blockingInvoke(){
        try {
            Map<String, Future<Object>> results = new LinkedHashMap<>();
            for (Map.Entry<String, Invoker> entry : invokers.entrySet()){
                Future<Object> future;
                if (syncThreadLocal) {
                    future = executor.submit(new ConcurrentCallableThreadLocal() {
                        @Override
                        public Object call() throws Exception {
                            setThreadLocal(threadLocalObj);
                            Object o = entry.getValue().invoke();
                            removeThreadLocal();
                            return o;
                        }
                    });
                } else {
                    future = executor.submit(() -> entry.getValue().invoke());
                }
                results.put(entry.getKey(), future);
            }
            Map<String, Object> returnMap = new LinkedHashMap<>();

            for (Map.Entry<String, Future<Object>> entry : results.entrySet()) {
                Future<Object> future = entry.getValue();
                try {
                    Object o = future.get(timeout, TimeUnit.MILLISECONDS);
                    returnMap.put(entry.getKey(), o);
                } catch (Exception e) {
                    throw new InvokerException("Failed to invoke service " +
                            entry.getKey() +
                            ": " +
                            e.getMessage(),
                            e);
                }
            }
            return returnMap;
        }finally {
            executor.shutdown();
        }
    }


    public static class Builder {

        private String name = "默认多线程任务提交器";
        private Map<String, Invoker> invokers = new LinkedHashMap<>();
        private long timeout = 5000;
        private boolean syncThreadLocal = false;
        private Object threadLocalObj;
        private int concurrentCount = Runtime.getRuntime().availableProcessors() << 1;

        public Builder() {
        }

        public Builder withConcurrentCount(int concurrentCount) {
            this.concurrentCount = concurrentCount;
            return this;
        }

        public Builder withInvoker(String code, Invoker invoker) {
            invokers.put(code, invoker);
            return this;
        }


        public Builder withInvoker(String code, ArgsInvoker invoker, Object... args) {
            invokers.put(code, () -> invoker.invoke(args));
            return this;
        }

        public Builder withTimeout(long timeout) {
            this.timeout = timeout;
            return this;
        }

        public Builder withThreadLocal(Object threadLocalObj) {
            this.syncThreadLocal = true;
            this.threadLocalObj = threadLocalObj;
            return this;
        }

        public Builder withName(String name){
            this.name=name;
            return this;
        }

        public ConcurrentInvoker build() {
            return new ConcurrentInvoker(name,this.invokers, this.timeout, syncThreadLocal, threadLocalObj, concurrentCount);
        }
    }


}
